import logging
import os
import time
from types import LambdaType
from typing import Any, Dict, List, Optional, Text, Tuple, Union

import numpy as np

from rasa.constants import DOCS_URL_POLICIES, DOCS_URL_DOMAINS
from rasa.core import jobs
from rasa.core.actions.action import (
    ACTION_LISTEN_NAME,
    ACTION_SESSION_START_NAME,
    Action,
    ActionExecutionRejection,
)
from rasa.core.channels.channel import (
    CollectingOutputChannel,
    OutputChannel,
    UserMessage,
)
from rasa.core.constants import (
    USER_INTENT_BACK,
    USER_INTENT_OUT_OF_SCOPE,
    USER_INTENT_RESTART,
    USER_INTENT_SESSION_START,
    UTTER_PREFIX,
)
from rasa.core.domain import Domain
from rasa.core.events import (
    ActionExecuted,
    ActionExecutionRejected,
    BotUttered,
    Event,
    ReminderCancelled,
    ReminderScheduled,
    SlotSet,
    UserUttered,
)
from rasa.core.interpreter import (
    INTENT_MESSAGE_PREFIX,
    NaturalLanguageInterpreter,
    RegexInterpreter,
)
from rasa.core.nlg import NaturalLanguageGenerator
from rasa.core.policies.ensemble import PolicyEnsemble
from rasa.core.tracker_store import TrackerStore
from rasa.core.trackers import DialogueStateTracker, EventVerbosity
from rasa.utils.common import raise_warning
from rasa.utils.endpoints import EndpointConfig


logger = logging.getLogger(__name__)
from rasa.constants import ENV_LOG_DEBUG
logger.addHandler(logging.StreamHandler())
logger.setLevel(ENV_LOG_DEBUG)

MAX_NUMBER_OF_PREDICTIONS = int(os.environ.get("MAX_NUMBER_OF_PREDICTIONS", "10"))

DEFAULT_INTENTS = [
    USER_INTENT_RESTART,
    USER_INTENT_BACK,
    USER_INTENT_OUT_OF_SCOPE,
    USER_INTENT_SESSION_START,
]


class MessageProcessor:
    """消息处理器"""

    def __init__(
        self,
        interpreter: NaturalLanguageInterpreter,
        policy_ensemble: PolicyEnsemble,
        domain: Domain,
        tracker_store: TrackerStore,
        generator: NaturalLanguageGenerator,
        action_endpoint: Optional[EndpointConfig] = None,
        max_number_of_predictions: int = MAX_NUMBER_OF_PREDICTIONS,
        message_preprocessor: Optional[LambdaType] = None,
        on_circuit_break: Optional[LambdaType] = None,
    ):
        self.interpreter = interpreter
        self.nlg = generator
        self.policy_ensemble = policy_ensemble
        self.domain = domain
        self.tracker_store = tracker_store
        self.max_number_of_predictions = max_number_of_predictions
        self.message_preprocessor = message_preprocessor
        self.on_circuit_break = on_circuit_break
        self.action_endpoint = action_endpoint

    async def handle_message(
        self, message: UserMessage
    ) -> Optional[List[Dict[Text, Any]]]:
        """使用此处理器处理单个消息：开始处理消息的"""
        logger.debug(f"\n ...........获取新的消息对象.......... \n {message.__dict__}")

        # （1）这里进行预处理：preprocess message if necessary
        tracker = await self.log_message(message, should_save_tracker=False)
        if not tracker:
            return None

        if not self.policy_ensemble or not self.domain:
            # 保存跟踪器状态以从此状态继续对话
            self._save_tracker(tracker)
            raise_warning(
                "No policy ensemble or domain set. Skipping action prediction "
                "and execution.",
                docs=DOCS_URL_POLICIES,
            )
            return None

        # （2）预测并执行下一步
        await self._predict_and_execute_next_action(message.output_channel, tracker)

        #  （3）保存跟踪器状态以从此状态继续对话
        self._save_tracker(tracker)

        if isinstance(message.output_channel, CollectingOutputChannel):
            return message.output_channel.messages
        else:
            return None

    async def predict_next(self, sender_id: Text) -> Optional[Dict[Text, Any]]:

        # we have a Tracker instance for each user
        # which maintains conversation state
        tracker = await self.get_tracker_with_session_start(sender_id)
        if not tracker:
            logger.warning(
                f"Failed to retrieve or create tracker for sender '{sender_id}'."
            )
            return None

        if not self.policy_ensemble or not self.domain:
            # save tracker state to continue conversation from this state
            raise_warning(
                "No policy ensemble or domain set. Skipping action prediction."
                "You should set a policy before training a model.",
                docs=DOCS_URL_POLICIES,
            )
            return None

        probabilities, policy = self._get_next_action_probabilities(tracker)
        # save tracker state to continue conversation from this state
        self._save_tracker(tracker)
        scores = [
            {"action": a, "score": p}
            for a, p in zip(self.domain.action_names, probabilities)
        ]
        return {
            "scores": scores,
            "policy": policy,
            "confidence": np.max(probabilities),
            "tracker": tracker.current_state(EventVerbosity.AFTER_RESTART),
        }

    async def _update_tracker_session(
        self, tracker: DialogueStateTracker, output_channel: OutputChannel
    ) -> None:
        """检查“跟踪器”中的当前会话并在过期时更新它。

         如果最新的跟踪器会话已过期，或者跟踪器尚未包含任何事件（仅考虑上次重启之后的事件），
         则运行“action_session_start”。

        Args:
            tracker: Tracker to inspect.
            output_channel: Output channel for potential utterances in a custom
                `ActionSessionStart`.
        """
        if not tracker.applied_events() or self._has_session_expired(tracker):
            logger.debug(
                f"Starting a new session for conversation ID '{tracker.sender_id}'."
            )

            await self._run_action(
                action=self._get_action(ACTION_SESSION_START_NAME),
                tracker=tracker,
                output_channel=output_channel,
                nlg=self.nlg,
            )

    async def get_tracker_with_session_start(
        self, sender_id: Text, output_channel: Optional[OutputChannel] = None
    ) -> Optional[DialogueStateTracker]:
        """获取 “sender_id” 的跟踪器，或为 “sender_id” 创建一个新的跟踪器。
        如果创建了新的跟踪器，则运行 `action_session_start`。这里的sender_id是区分不同用户的id

        Args:
            output_channel: 与传入的用户消息关联的输出通道。
            sender_id: 要为其获取跟踪器的对话 ID。
        Returns:
              如果可用，则为 `sender_id` 的跟踪器，否则为 `None`。
        """
        # 获取sender_id的跟踪器
        tracker = self.get_tracker(sender_id)
        if not tracker:
            return None
        # 更新跟踪器会话
        await self._update_tracker_session(tracker, output_channel)

        return tracker

    def get_tracker(self, conversation_id: Text) -> Optional[DialogueStateTracker]:
        """获取对话跟踪器

        与 `get_tracker_with_session_start` 相比，
        这不会在对话开始时添加任何 `action_session_start` 或 `session_start` 事件。

        参数：session_id：应检索其历史记录的对话的 ID。
        返回：对话跟踪器。 如果是新对话，则创建一个空的跟踪器。
        """
        conversation_id = conversation_id or UserMessage.DEFAULT_SENDER_ID
        return self.tracker_store.get_or_create_tracker(
            conversation_id, append_action_listen=False
        )

    async def log_message(
        self, message: UserMessage, should_save_tracker: bool = True
    ) -> Optional[DialogueStateTracker]:
        """在属于消息会话 ID 的跟踪器上记录“消息”。
        如果 `should_save_tracker` 为 `True`，则可选择保存跟踪器。
        如果此方法返回的跟踪器用于进一步处理并在稍后阶段保存，则可以跳过跟踪器保存。
        """

        # 如有必要，预处理消息
        if self.message_preprocessor is not None:
            message.text = self.message_preprocessor(message.text)

        # 我们为每个用户都有一个 Tracker 实例，用于维护对话状态
        tracker = await self.get_tracker_with_session_start(
            message.sender_id, message.output_channel
        )
        # 使用跟踪器，开始处理消息
        if tracker:
            await self._handle_message_with_tracker(message, tracker)
            if should_save_tracker:
                self._save_tracker(tracker)   # 保存跟踪器状态以从此状态继续对话
        else:
            logger.warning(
                f"Failed to retrieve or create tracker for conversation ID "
                f"'{message.sender_id}'."
            )
        return tracker

    async def execute_action(
        self,
        sender_id: Text,
        action_name: Text,
        output_channel: OutputChannel,
        nlg: NaturalLanguageGenerator,
        policy: Text,
        confidence: float,
    ) -> Optional[DialogueStateTracker]:

        # we have a Tracker instance for each user
        # which maintains conversation state
        tracker = await self.get_tracker_with_session_start(sender_id, output_channel)
        if tracker:
            action = self._get_action(action_name)
            await self._run_action(
                action, tracker, output_channel, nlg, policy, confidence
            )

            # save tracker state to continue conversation from this state
            self._save_tracker(tracker)
        else:
            logger.warning(
                f"Failed to retrieve or create tracker for conversation ID "
                f"'{sender_id}'."
            )
        return tracker

    def predict_next_action(
        self, tracker: DialogueStateTracker
    ) -> Tuple[Action, Text, float]:
        """预测机器人在看到 x 后应该采取的下一步行动。
        这应该被更高级的策略覆盖，以使用 ML 来预测动作。 返回下一个操作的索引。"""

        action_confidences, policy = self._get_next_action_probabilities(tracker)

        max_confidence_index = int(np.argmax(action_confidences))
        action = self.domain.action_for_index(
            max_confidence_index, self.action_endpoint
        )
        logger.debug(
            f"Predicted next action '{action.name()}' with confidence "
            f"{action_confidences[max_confidence_index]:.2f}."
        )
        return action, policy, action_confidences[max_confidence_index]

    @staticmethod
    def _is_reminder(e: Event, name: Text) -> bool:
        return isinstance(e, ReminderScheduled) and e.name == name

    @staticmethod
    def _is_reminder_still_valid(
        tracker: DialogueStateTracker, reminder_event: ReminderScheduled
    ) -> bool:
        """Check if the conversation has been restarted after reminder."""

        for e in reversed(tracker.applied_events()):
            if MessageProcessor._is_reminder(e, reminder_event.name):
                return True
        return False  # not found in applied events --> has been restarted

    @staticmethod
    def _has_message_after_reminder(
        tracker: DialogueStateTracker, reminder_event: ReminderScheduled
    ) -> bool:
        """Check if the user sent a message after the reminder."""

        for e in reversed(tracker.events):
            if MessageProcessor._is_reminder(e, reminder_event.name):
                return False
            elif isinstance(e, UserUttered) and e.text:
                return True
        return True  # tracker has probably been restarted

    async def handle_reminder(
        self,
        reminder_event: ReminderScheduled,
        sender_id: Text,
        output_channel: OutputChannel,
        nlg: NaturalLanguageGenerator,
    ) -> None:
        """Handle a reminder that is triggered asynchronously."""

        tracker = await self.get_tracker_with_session_start(sender_id, output_channel)

        if not tracker:
            logger.warning(
                f"Failed to retrieve tracker for conversation ID '{sender_id}'."
            )
            return None

        if (
            reminder_event.kill_on_user_message
            and self._has_message_after_reminder(tracker, reminder_event)
            or not self._is_reminder_still_valid(tracker, reminder_event)
        ):
            logger.debug(
                f"Canceled reminder because it is outdated ({reminder_event})."
            )
        else:
            intent = reminder_event.intent
            entities = reminder_event.entities or {}
            await self.trigger_external_user_uttered(
                intent, entities, tracker, output_channel
            )

    async def trigger_external_user_uttered(
        self,
        intent_name: Text,
        entities: Optional[Union[List[Dict[Text, Any]], Dict[Text, Text]]],
        tracker: DialogueStateTracker,
        output_channel: OutputChannel,
    ) -> None:
        """Triggers an external message.

        Triggers an external message (like a user message, but invisible;
        used, e.g., by a reminder or the trigger_intent endpoint).

        Args:
            intent_name: Name of the intent to be triggered.
            entities: Entities to be passed on.
            tracker: The tracker to which the event should be added.
            output_channel: The output channel.
        """
        if isinstance(entities, list):
            entity_list = entities
        elif isinstance(entities, dict):
            # Allow for a short-hand notation {"ent1": "val1", "ent2": "val2", ...}.
            # Useful if properties like 'start', 'end', or 'extractor' are not given,
            # e.g. for external events.
            entity_list = [
                {"entity": ent, "value": val} for ent, val in entities.items()
            ]
        elif not entities:
            entity_list = []
        else:
            raise_warning(
                f"Invalid entity specification: {entities}. Assuming no entities."
            )
            entity_list = []
        tracker.update(UserUttered.create_external(intent_name, entity_list))
        await self._predict_and_execute_next_action(output_channel, tracker)
        # save tracker state to continue conversation from this state
        self._save_tracker(tracker)

    @staticmethod
    def _log_slots(tracker) -> None:
        """记录当前设置的插槽"""

        slot_values = "\n".join(
            [f"\t{s.name}: {s.value}" for s in tracker.slots.values()]
        )
        if slot_values.strip():
            logger.debug(f"Current slot values: \n{slot_values}")

    def _log_unseen_features(self, parse_data: Dict[Text, Any]) -> None:
        """检查 NLU 解释器是否选择了无法识别的意图和实体。"""

        domain_is_not_empty = self.domain and not self.domain.is_empty()

        intent = parse_data["intent"]["name"]
        if intent:
            intent_is_recognized = (
                domain_is_not_empty and intent in self.domain.intents
            ) or intent in DEFAULT_INTENTS
            if not intent_is_recognized:
                raise_warning(
                    f"Interpreter parsed an intent '{intent}' "
                    f"which is not defined in the domain. "
                    f"Please make sure all intents are listed in the domain.",
                    docs=DOCS_URL_DOMAINS,
                )

        entities = parse_data["entities"] or []
        for element in entities:
            entity = element["entity"]
            if entity and domain_is_not_empty and entity not in self.domain.entities:
                raise_warning(
                    f"Interpreter parsed an entity '{entity}' "
                    f"which is not defined in the domain. "
                    f"Please make sure all entities are listed in the domain.",
                    docs=DOCS_URL_DOMAINS,
                )

    def _get_action(self, action_name) -> Optional[Action]:
        return self.domain.action_for_name(action_name, self.action_endpoint)

    async def _parse_message(self, message, tracker: DialogueStateTracker = None):
        """
        用于测试 - 您可以使用 /intent{"entity1": val1, "entity2": val2} 格式的消息来缩短 NLU 部分
        parse_data is a dict of intent & entities
        """

        if message.text.startswith(INTENT_MESSAGE_PREFIX):
            parse_data = await RegexInterpreter().parse(
                message.text, message.message_id, tracker
            )
        else:
            parse_data = await self.interpreter.parse(
                message.text, message.message_id, tracker
            )

        logger.debug(
            "Received user message '{}' with intent '{}' "
            "and entities '{}'".format(
                message.text, parse_data["intent"], parse_data["entities"]
            )
        )

        self._log_unseen_features(parse_data)

        return parse_data

    async def _handle_message_with_tracker(
        self, message: UserMessage, tracker: DialogueStateTracker
    ) -> None:

        if message.parse_data:
            parse_data = message.parse_data
        else:
            parse_data = await self._parse_message(message, tracker)

        # 永远不要直接改变跟踪器——而是将其事件传递到日志
        tracker.update(
            UserUttered(
                message.text,
                parse_data["intent"],
                parse_data["entities"],
                parse_data,
                input_channel=message.input_channel,
                message_id=message.message_id,
                metadata=message.metadata,
            ),
            self.domain,
        )

        if parse_data["entities"]:
            self._log_slots(tracker)

        logger.debug(
            f"Logged UserUtterance - tracker now has {len(tracker.events)} events."
        )

    @staticmethod
    def _should_handle_message(tracker: DialogueStateTracker):
        return (
            not tracker.is_paused()
            or tracker.latest_message.intent.get("name") == USER_INTENT_RESTART
        )

    async def _predict_and_execute_next_action(
        self, output_channel: OutputChannel, tracker: DialogueStateTracker
    ):
        """采取策略决定的 action，直到它选择 “listen” """

        should_predict_another_action = True
        num_predicted_actions = 0

        def is_action_limit_reached():
            return (
                num_predicted_actions == self.max_number_of_predictions
                and should_predict_another_action
            )

        # action loop：环形预测动作，直到我们点击动作监听
        while (
            should_predict_another_action
            and self._should_handle_message(tracker)
            and num_predicted_actions < self.max_number_of_predictions
        ):
            # 这实际上只是以相同的名称调用策略的方法
            action, policy, confidence = self.predict_next_action(tracker)

            should_predict_another_action = await self._run_action(
                action, tracker, output_channel, self.nlg, policy, confidence
            )
            num_predicted_actions += 1

        if is_action_limit_reached():
            # circuit breaker was tripped
            logger.warning(
                "Circuit breaker tripped. Stopped predicting "
                f"more actions for sender '{tracker.sender_id}'."
            )
            if self.on_circuit_break:
                self.on_circuit_break(tracker, output_channel, self.nlg)  # 调用已注册的回调

    @staticmethod
    def should_predict_another_action(action_name: Text) -> bool:
        """确定处理器是否应该预测另一个动作

        Args:
            action_name: Name of the latest executed action.

        Returns:
            `False` if `action_name` is `ACTION_LISTEN_NAME` or
            `ACTION_SESSION_START_NAME`, otherwise `True`.
        """

        return action_name not in (ACTION_LISTEN_NAME, ACTION_SESSION_START_NAME)

    @staticmethod
    async def _send_bot_messages(
        events: List[Event],
        tracker: DialogueStateTracker,
        output_channel: OutputChannel,
    ) -> None:
        """发送记录在事件数组中的所有机器人消息：通过输出通道发送消息的地方"""

        for e in events:
            if not isinstance(e, BotUttered):
                continue

            await output_channel.send_response(tracker.sender_id, e.message())

    async def _schedule_reminders(
        self,
        events: List[Event],
        tracker: DialogueStateTracker,
        output_channel: OutputChannel,
        nlg: NaturalLanguageGenerator,
    ) -> None:
        """使用调度程序为作业计时以触发通过的提醒。

        具有相同 `id` 属性的提醒将相互覆盖（即最终只有其中一个会运行）.
        """

        for e in events:
            if not isinstance(e, ReminderScheduled):
                continue

            (await jobs.scheduler()).add_job(
                self.handle_reminder,
                "date",
                run_date=e.trigger_date_time,
                args=[e, tracker.sender_id, output_channel, nlg],
                id=e.name,
                replace_existing=True,
                name=e.scheduled_job_name(tracker.sender_id),
            )

    @staticmethod
    async def _cancel_reminders(
        events: List[Event], tracker: DialogueStateTracker
    ) -> None:
        """取消与“ReminderCancelled”事件匹配的提醒。"""

        # ReminderCancelled 事件指定的所有提醒将被取消
        for event in events:
            if isinstance(event, ReminderCancelled):
                scheduler = await jobs.scheduler()
                for scheduled_job in scheduler.get_jobs():
                    if event.cancels_job_with_name(
                        scheduled_job.name, tracker.sender_id
                    ):
                        scheduler.remove_job(scheduled_job.id)

    async def _run_action(
        self, action, tracker, output_channel, nlg, policy=None, confidence=None
    ) -> bool:
        # 事件和返回值用于在采取行动后
        # 更新跟踪器状态
        try:
            events = await action.run(output_channel, nlg, tracker, self.domain)
        except ActionExecutionRejection:
            events = [ActionExecutionRejected(action.name(), policy, confidence)]
            tracker.update(events[0])
            return self.should_predict_another_action(action.name())
        except Exception as e:
            logger.error(
                f"Encountered an exception while running action '{action.name()}'. "
                "Bot will continue, but the actions events are lost. "
                "Please check the logs of your action server for "
                "more information."
            )
            logger.debug(e, exc_info=True)
            events = []

        self._log_action_on_tracker(tracker, action.name(), events, policy, confidence)
        if action.name() != ACTION_LISTEN_NAME and not action.name().startswith(
            UTTER_PREFIX
        ):
            self._log_slots(tracker)
        # 发动消息、调度器
        await self._send_bot_messages(events, tracker, output_channel)
        await self._schedule_reminders(events, tracker, output_channel, nlg)
        await self._cancel_reminders(events, tracker)

        return self.should_predict_another_action(action.name())

    def _warn_about_new_slots(self, tracker, action_name, events) -> None:
        """这些是我们在训练期间看到的那个动作的事件"""

        if (
            not self.policy_ensemble
            or action_name not in self.policy_ensemble.action_fingerprints
        ):
            return

        fp = self.policy_ensemble.action_fingerprints[action_name]
        slots_seen_during_train = fp.get("slots", set())
        for e in events:
            if isinstance(e, SlotSet) and e.key not in slots_seen_during_train:
                s = tracker.slots.get(e.key)
                if s and s.has_features():
                    if e.key == "requested_slot" and tracker.active_form:
                        pass
                    else:
                        raise_warning(
                            f"Action '{action_name}' set a slot type '{e.key}' which "
                            f"it never set during the training. This "
                            f"can throw off the prediction. Make sure to "
                            f"include training examples in your stories "
                            f"for the different types of slots this "
                            f"action can return. Remember: you need to "
                            f"set the slots manually in the stories by "
                            f"adding '- slot{{\"{e.key}\": {e.value}}}' "
                            f"after the action."
                        )

    def _log_action_on_tracker(
        self, tracker, action_name, events, policy, confidence
    ) -> None:
        # 确保即使懒惰的程序员在操作结束时
        # 未键入 `return []` 或 run 方法由于某些其他原因返回 `None`，
        # 代码仍然有效。
        if events is None:
            events = []

        logger.debug(
            f"Action '{action_name}' ended with events '{[e for e in events]}'."
        )

        self._warn_about_new_slots(tracker, action_name, events)

        if action_name is not None:
            # 记录操作及其产生的事件
            tracker.update(ActionExecuted(action_name, policy, confidence))

        for e in events:
            # 这确保事件按时间戳排序。
            #
            # 由于事件对象是在其他地方创建的，
            # 时间戳将指示执行操作的时间之前的时间
            e.timestamp = time.time()
            tracker.update(e, self.domain)

    def _has_session_expired(self, tracker: DialogueStateTracker) -> bool:
        """Determine whether the latest session in `tracker` has expired.

        Args:
            tracker: Tracker to inspect.

        Returns:
            `True` if the session in `tracker` has expired, `False` otherwise.
        """

        if not self.domain.session_config.are_sessions_enabled():
            # tracker has never expired if sessions are disabled
            return False

        user_uttered_event: Optional[UserUttered] = tracker.get_last_event_for(
            UserUttered
        )

        if not user_uttered_event:
            # there is no user event so far so the session should not be considered
            # expired
            return False

        time_delta_in_seconds = time.time() - user_uttered_event.timestamp
        has_expired = (
            time_delta_in_seconds / 60
            > self.domain.session_config.session_expiration_time
        )
        if has_expired:
            logger.debug(
                f"The latest session for conversation ID '{tracker.sender_id}' has "
                f"expired."
            )

        return has_expired

    def _save_tracker(self, tracker: DialogueStateTracker) -> None:
        self.tracker_store.save(tracker)

    def _prob_array_for_action(
        self, action_name: Text
    ) -> Tuple[Optional[List[float]], None]:
        """action的概率-数组"""

        idx = self.domain.index_for_action(action_name)
        if idx is not None:
            result = [0.0] * self.domain.num_actions
            result[idx] = 1.0
            return result, None
        else:
            return None, None

    def _get_next_action_probabilities(
        self, tracker: DialogueStateTracker
    ) -> Tuple[Optional[List[float]], Optional[Text]]:
        """从集合策略中收集预测并返回动作和预测。"""

        followup_action = tracker.followup_action
        if followup_action:
            tracker.clear_followup_action()  # 清除
            result = self._prob_array_for_action(followup_action)
            if result:
                return result
            else:
                logger.error(
                    f"Trying to run unknown follow-up action '{followup_action}'!"
                    "Instead of running that, we will ignore the action "
                    "and predict the next action."
                )

        return self.policy_ensemble.probabilities_using_best_policy(
            tracker, self.domain
        )
